package pers.cz.thread;

import java.util.HashMap;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedDeque;

/**
 * @program: PostGirl-panent
 * @description: Master-Worker模式: Master主要用于任务分发和汇总
 * @author: Cheng Zhi
 * @create: 2022-05-03 22:17
 **/
public class Master {

    // 任务队列
    protected Queue<Object> workQueue = new ConcurrentLinkedDeque<>();
    // 分表编号
    protected Queue<Integer> shardingQueue = new ConcurrentLinkedDeque<>();
    // worker线程队列
    protected Map<String, Worker> threadMap = new HashMap<String, Worker>(); // 拟采用线程池的方式，但是线程池停止不及时
    //protected Map<String, Thread> threadMap = new HashMap<String, Thread>();
    // 子任务处理结果集
    protected Map<String, Object> resultMap = new ConcurrentHashMap<>();

    PartitionQueryThreadExecutor partitionQueryThreadExecutor;

    /**
     * 构造Master
     * @param worker
     * @param countWorker
     */
    public Master(Worker worker, int countWorker) {
        worker.setWorkQueue(workQueue);
        worker.setResultMap(resultMap);
        worker.setShardingQueue(shardingQueue);
        partitionQueryThreadExecutor = PartitionQueryThreadExecutor.getInstance(0, countWorker, countWorker + 5);
        partitionQueryThreadExecutor.start();
        for (int i=0; i<countWorker; i++) {
            //threadMap.put(Integer.toString(i), new Thread(worker, Integer.toString(i)));
            threadMap.put(Integer.toString(i), worker);
        }
    }

    /**
     * 提交一个任务
     * @param job
     */
    public void submit(Object job, Integer shardingNum) {
        workQueue.add(job);
        shardingQueue.add(shardingNum);
    }

    /**
     * 获取结果集
     * @return
     */
    public Map<String, Object> getResultMap() {
        return resultMap;
    }

    /**
     * 开始运行所有的worker任务
     */
    public void execute() {

        for (Map.Entry<String, Worker> entry : threadMap.entrySet()) {
            //entry.getValue().start();
            partitionQueryThreadExecutor.executeTask(entry.getValue());
        }
    }

    /**
     * 通知线程池停止
     */
    public void stop() {
        partitionQueryThreadExecutor.stop();
    }

    /**
     * 判断工作线程是否完全停止
     * @return
     */
    public boolean isComplete() {

/*        for (Map.Entry entry : threadMap.entrySet()) {

            Thread thread = (Thread) entry.getValue();
            if (thread.getState() != Thread.State.TERMINATED) {
                return false;
            }
        }
        return true;*/
        //return partitionQueryThreadExecutor.isProductThreadStop();
        return partitionQueryThreadExecutor.isAllStop();
    }




}
